
package com.leo.jdkcore.kafka;

import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

/**
 * ktype: key 
 * vtype: value 
 * @param <Vtype>
 */
public class KafkaMsgProducer<Ktype,Vtype> {

	private final KafkaProducer<Integer, Vtype> producer;
	private final String topic;
	private AtomicInteger messageNo = new AtomicInteger(0);
	
	public String getTopic() {
		return topic;
	}

	public KafkaMsgProducer(String topic,Properties props,Class<Vtype> vTypeclazz,Class<Ktype> kTypeclazz) {
		
		if(kTypeclazz.getName().equalsIgnoreCase(String.class.getName())){
			props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
		}else if (kTypeclazz.getName().equalsIgnoreCase(Integer.class.getName())){
			props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.IntegerDeserializer");
		}
		
		if(vTypeclazz.getName().equalsIgnoreCase(String.class.getName())){
			props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		}else{
			props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
		}
		producer = new KafkaProducer<Integer, Vtype>(props);
		this.topic = topic;
	}

	/**
	 * @param data
	 */
	public void sendData(Vtype data) {
		try {
			_sendData(data);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * @param data
	 */
	public void sendDataAsync(Vtype data, Callback callback) {
		try {
			_sendDataAsync(data, callback);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}


	/**
	 * @param data
	 */
	private void _sendData(Vtype data) throws Exception {
		producer.send(
				new ProducerRecord<Integer, Vtype>(topic, getMessageNo(), data)).get();
		
		//System.err.println("producer data ok,partation:"+metadata.partition() );
	}

	/**
	 * @param data
	 * @param callback
	 */
	private void _sendDataAsync(Vtype data, Callback callback) throws Exception {
		producer.send(new ProducerRecord<Integer, Vtype>(topic,getMessageNo(), data), callback);
	}
	
	private int getMessageNo(){
		if (messageNo.getAndIncrement() == Integer.MAX_VALUE) {
			messageNo.set(0);
		}
		return messageNo.get();
	}
}